package cn.sczc.jh.client.service.synch;

import java.io.IOException;
import java.text.NumberFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.alibaba.fastjson.JSONObject;
import com.google.common.util.concurrent.AbstractExecutionThreadService;

import cn.sczc.jh.client.entity.SensorHistory;
import cn.sczc.jh.client.hbase.HbaseConn;
import cn.sczc.jh.client.hbase.HbaseUtils;
import cn.sczc.jh.client.mapper.sensorHistroy.SensorHistoryMapper;
import cn.sczc.jh.client.vo.SynchMsgVo;

public class SynchExecutionThreadServiceImpl extends AbstractExecutionThreadService {
	private final static Logger logger = LogManager.getLogger(SynchExecutionThreadServiceImpl.class);
	private SensorHistoryMapper sensorHistroyMapper;
	private Map<String, Object> params;
	private SynchMsgVo synchMsgVo;
	private SimpleDateFormat sdf;

	public SynchExecutionThreadServiceImpl(SensorHistoryMapper sensorHistroyMapper, Map<String, Object> params) {
		this.sensorHistroyMapper = sensorHistroyMapper;
		this.params = params;
		synchMsgVo = new SynchMsgVo();
		sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	}

	@Override
	protected void startUp() throws IOException {
		HbaseUtils.createTable("sensor_data", new String[] { "base_info", "property_info" });
	}

	@Override
	protected void run() {
		try {
			int total = sensorHistroyMapper.getSynchSensorDataCount(params);
			logger.info("共计----------" + total);

			if (total == 0) {
				synchMsgVo.setErrMsg("暂无可同步的数据！");
				SynchWebsocket.BroadCastInfo(JSONObject.toJSONString(synchMsgVo));
				return;
			}
			int pagenNum = 5000;
			params.put("pagenNum", pagenNum);
			int synchNum = 0;
			long startTime = System.currentTimeMillis();
			synchMsgVo.setTotal(total);
			synchMsgVo.setPercent("0");
			SynchWebsocket.BroadCastInfo(JSONObject.toJSONString(synchMsgVo));
			while (total > synchNum) {
				logger.info("-----开始查询-----");
				long startQuery = System.currentTimeMillis();
				List<SensorHistory> sensorHistoryList = sensorHistroyMapper.getSynchSensorData(params);
				logger.info("-----查询完成-----用时:" + (System.currentTimeMillis() - startQuery) / 1000.0 + "s");
				if (sensorHistoryList == null || sensorHistoryList.size() == 0) {
					return;
				}
				List<Put> puts = new ArrayList<>();
				List<String> ids = new ArrayList<>();
				for (SensorHistory sensorHistory : sensorHistoryList) {
					Put put = this.getPut(sensorHistory);
					puts.add(put);
					ids.add(sensorHistory.getOid());
				}
				logger.info("-----开始同步-----");
				long startSynch = System.currentTimeMillis();
				boolean isSynch = HbaseUtils.putRows("sensor_data", puts);
				logger.info("-----同步完成-----用时:" + (System.currentTimeMillis() - startSynch) / 1000.0 + "s");
				if (isSynch) {
					logger.info("-----开始更新-----");
					long startupdate = System.currentTimeMillis();
					sensorHistroyMapper.updateUploadStateByIds(ids);
					logger.info("-----更新完成-----用时:" + (System.currentTimeMillis() - startupdate) / 1000.0 + "s");
				} else {
					// 同步数据失败！
					synchMsgVo.setErrMsg("同步错误，请重试！");
					SynchWebsocket.BroadCastInfo(JSONObject.toJSONString(synchMsgVo));
					return;
				}
				synchNum += sensorHistoryList.size();
				logger.info("-----已同步-----" + synchNum);

				synchMsgVo.setSend(synchNum);
				synchMsgVo.setPercent(generatePercent(synchNum, total));
				SynchWebsocket.BroadCastInfo(JSONObject.toJSONString(synchMsgVo));
				Thread.sleep(50);
			}
			long endtime = System.currentTimeMillis();
			logger.info("共计用时" + (endtime - startTime) / 1000.0 + "s");

		} catch (Exception e) {
			logger.info(e.getMessage());
		}
	}

	/**
	 * 生成百分比
	 * 
	 * @param num
	 * @param total
	 * @return
	 */
	private String generatePercent(int num, int total) {
		// 创建一个数值格式化对象
		NumberFormat numberFormat = NumberFormat.getInstance();
		// 设置精确到小数点后2位
		numberFormat.setMaximumFractionDigits(2);
		String result = numberFormat.format((float) num / (float) total * 100);
		logger.info("百分比==" + result);
		return result;
	}

	private Put getPut(SensorHistory sensorHistory) {
		String rowkey = UUID.randomUUID().toString();
		Put put = new Put(Bytes.toBytes(rowkey));
		put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("equ_no"), Bytes.toBytes(sensorHistory.getEquNo()));
		put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sensor_address"),
				Bytes.toBytes(sensorHistory.getSensorAddress()));
		put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("create_time"),
				Bytes.toBytes(sdf.format(sensorHistory.getCreateTime())));
		Map<String, Object> propertyMap = JSONObject.parseObject(sensorHistory.getProperties());
		if (propertyMap != null && propertyMap.size() > 0) {
			for (Map.Entry<String, Object> entry : propertyMap.entrySet()) {
				put.addColumn(Bytes.toBytes("property_info"), Bytes.toBytes(entry.getKey()),
						Bytes.toBytes(String.valueOf(entry.getValue())));
			}
		}
		return put;
	}

	@Override
	protected void triggerShutdown() {
		HbaseConn.closeConn();
	}

	@Override
	protected void shutDown() throws Exception {
		HbaseConn.closeConn();
	}
}
